//
// Created by dongbo on 11/11/18.
//

#include "role_node.h"
#include "../z_hglog.h"
#include "evpaxos.h"
#include "paxos.h"


#include <iostream>
#include <stdio.h>
#include <errno.h>
#include <stdlib.h>
#include <time.h>
#include <string.h>
#include <signal.h>

#include <event2/event.h>
#include <netinet/tcp.h>

#include <atomic>
#include <mutex>
#include <thread>
#include <sys/prctl.h>
#include <signal.h>
#include <unistd.h>
#include <assert.h>


std::atomic_int role::master_id = {-1};
std::atomic_int role::lease_timep = {0};
int role::id = -2;
struct evpaxos_config *role::config = nullptr;


role_learner::role_learner(int _id,
                           role_acceptor *accp,
                           role_proposer *prop,
                           ppax_cb *_ppxcb,
                           struct evpaxos_config *_config) : zmqpip_num(0), ok(false), raccp(accp), rprop(prop) {
    id = _id;
    config = _config;
    ppxcb = _ppxcb;
}

void role_learner::handle_sigint(int sig, short ev, void *arg) {
    struct event_base *base = reinterpret_cast<event_base *>(arg);
    printf("Caught signal %d\n", sig);
    event_base_loopexit(base, NULL);
}

void role_learner::deliver(unsigned iid, char *value, size_t size, void *arg) {
    struct client_value *val = (struct client_value *) value;
    char tmp[4096] = {0};
    int now_timp = time(0);

    static std::map<std::string, std::set<int>> topiclist;
    static unsigned long long id = 0;

    if (arg != nullptr) {
        auto r_learner = (role_learner *) arg;
        r_learner->ok = true;

        std::istringstream iss(std::string(val->value, val->size));
        std::cout << iss.str() << std::endl;
        T_LOGI("recv:" << iss.str())
        int xyid = -1;
        iss >> xyid;

        if (xyid == t_paxos_msg::lease_msg) {
            t_paxos_msg::lease tp;
            tp.to_struct(iss);
            if (tp.time_p < now_timp) return; // for reboot slave node

            if (now_timp > r_learner->get_lease_timep()) { //new req to be master
                r_learner->set_lease_timep(tp.time_p);
                r_learner->master_id = tp.id;
                r_learner->event_master_change();
                r_learner->raccp->event_master_change();
                r_learner->rprop->event_master_change();
            } else if (r_learner->is_master(tp.id)) { //recve lease timeppint
                r_learner->set_lease_timep(tp.time_p);
                r_learner->master_id = tp.id;
            }
        } else if (xyid == t_paxos_msg::distribute_msg) {
            t_paxos_msg::distribute tp;
            tp.to_struct(iss);

            if (tp.id == r_learner->id) {
                topiclist[tp.strTopic].insert(tp.topic_gid);
                T_LOGI("distribute_topic:" << tp.id << "," << tp.strTopic << "," << tp.topic_gid);
            }
        } else if (xyid == t_paxos_msg::ordinary_msg) {
            t_paxos_msg::ordinary tp;
            tp.to_struct(iss);

            auto it = topiclist[tp.strTopic].find(tp.nGroupID);
            if (it != topiclist[tp.strTopic].end()) {
                auto job = new JOB();
                job->areaid = tp.nAreaID;
                job->groupid = tp.nGroupID;
                snprintf(job->topic, sizeof(job->topic), "%s", tp.strTopic.c_str());
                snprintf(job->tabname, sizeof(job->tabname), "%s", tp.tableName.c_str());
                job->slen = tp.content_len;
                snprintf(job->content, sizeof(job->content), "%s", tp.strContent.c_str());

                if (r_learner->zmqpip_num > 0) { //xyid groudpid areaid groupid topic tbname content
                    auto &zmqpip = r_learner->zpip_in[id++ % r_learner->zmqpip_num];
                    auto addr_ll = (unsigned long long) job;
                    zmqpip.sneddata((char *) &addr_ll, sizeof(unsigned long long));
                    T_LOGI("zpip_num:" << r_learner->zmqpip_num);
                }
            }
        }
    }
}

void role_learner::start_learner() {
    new std::thread([this]() {
        prctl(PR_SET_NAME, "learner");
        static std::mutex mux;
        struct evlearner *lea;
        struct event_base *base;
        {
            std::lock_guard<std::mutex> lock(mux);
            base = event_base_new();
            lea = evlearner_init(nullptr, deliver, this, base, reinterpret_cast<struct evpaxos_config *>(config));
            if (lea == NULL) {
                T_LOGW("Could not start the learner!");
                assert(lea != NULL);
                return;
            }
        }
//        {
//            sigset_t sset;
//            sigemptyset(&sset);
//            sigaddset(&sset,SIGPIPE);
//            pthread_sigmask(SIG_BLOCK, &sset, NULL);
//        }

        event_base_dispatch(base);
        evlearner_free(lea);
        event_base_free(base);
    });
}


void role_learner::set_zpip(zmqpipe *_zpip_in, int _zmq_num) {
    zmqpip_num = _zmq_num;
    zpip_in = _zpip_in;
    T_LOGI("zmqpip_num:" << zmqpip_num);
}

void role_learner::set_masterid(int _id) {
    master_id = _id;
}

void role_learner::set_lease_timep(int timep) {
    lease_timep = timep;
}

int role_learner::get_masterid() {
    return master_id;
}

bool role_learner::is_master(int _id) {
    return master_id == _id;
}

void role_learner::event_master_change() {

}

int role_learner::get_lease_timep() {
    return lease_timep;
}

role_acceptor::role_acceptor(ppax_cb *_ppxcb) {
    ppxcb = _ppxcb;
}

void role_acceptor::handle_sigint(int sig, short ev, void *arg) {
    struct event_base *base = reinterpret_cast<event_base *>(arg);
    printf("Caught signal %d\n", sig);
    event_base_loopexit(base, NULL);
}

void role_acceptor::start_acceptor() {
    new std::thread([this]() {
        prctl(PR_SET_NAME, "acceptor"); //#include <sys/prctl.h>
        static std::mutex mux;
        struct evacceptor *acc;
        struct event_base *base;
        {
            std::lock_guard<std::mutex> lock(mux);
            base = event_base_new();
            acc = evacceptor_init(id, nullptr, base, reinterpret_cast<struct evpaxos_config *>(config));
            if (acc == NULL) {
                T_LOGW("Could not start the acceptor!");
                assert(acc != NULL);
                return;
            }
        }

//        {
//            sigset_t sset;
//            sigemptyset(&sset);
//            sigaddset(&sset,SIGPIPE);
//            pthread_sigmask(SIG_BLOCK, &sset, NULL);
//        }

        event_base_dispatch(base);
        evacceptor_free(acc);
        event_base_free(base);
    });
}

void role_acceptor::set_masterid(int _id) {
    master_id = _id;
}

void role_acceptor::set_lease_timep(int timep) {
    lease_timep = timep;
}

int role_acceptor::get_masterid() {
    return master_id;
}

bool role_acceptor::is_master(int _id) {
    return master_id == _id;
}

void role_acceptor::event_master_change() {

}

int role_acceptor::get_lease_timep() {
    return lease_timep;
}

role_proposer::role_proposer(ppax_cb *_ppxcb) {
    ppxcb = _ppxcb;
}

void role_proposer::handle_sigint(int sig, short ev, void *arg) {
    struct event_base *base = reinterpret_cast<event_base *>(arg);
    printf("Caught signal %d\n", sig);
    event_base_loopexit(base, NULL);
}

void role_proposer::start_proposer() {
    new std::thread([this]() {
        prctl(PR_SET_NAME, "proposer");
        static std::mutex mux;
        struct event_base *base;
        struct evproposer *prop;
        {
            std::lock_guard<std::mutex> lock(mux);
            base = event_base_new();
            prop = evproposer_init(id,
                                   this,
                                   base,
                                   reinterpret_cast<struct evpaxos_config *>(config));
            if (prop == NULL) {
                T_LOGW("Could not start the proposer!");
                assert(prop != NULL);
                return;
            }
        }

        event_base_dispatch(base);
        evproposer_free(prop);
        event_base_free(base);
    });
}

void role_proposer::set_masterid(int _id) {
    master_id = _id;
}

void role_proposer::set_lease_timep(int timep) {
    lease_timep = timep;
}

int role_proposer::get_masterid() {
    return master_id;
}

bool role_proposer::is_master(int _id) {
    return master_id == _id;
}

void role_proposer::event_master_change() {
    T_LOGW("master is change, nowmaster is:" << get_masterid());
    if (ppxcb) {
        if (get_masterid() == id)
            ppxcb->master_ischange();
    }
}

int role_proposer::get_lease_timep() {
    return lease_timep;
}